package com.atguigu.gmall.realtime.dwd.app;

/*import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TableProcessDwd;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import com.google.gson.JsonObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

*//**
 * DWD事务事实表:
 *      流量域
 *          未经加工的事务事实表 -> 日志分流
 *
 *      交易域
 *          加购事务事实表
 *          下单事务事实表
 *          取消订单事务事实表
 *          支付成功事务事实表
 *          退单事务事实表
 *          退款成功事务事实表
 *
 *      互动域
 *          评论事务事实表
 *          收藏商品事务事实表   favor_info表中 -> 筛选insert
 *
 *      工具域
 *          优惠券领取事务事实表  coupon_use表中 -> 筛选insert数据
 *          优惠券使用事务事实表  coupon_use表中 -> 筛选update数据
 *
 *      用户域
 *          用户注册事务事实表   user_info表中 -> 筛选insert数据
 *
 *
 *//*
public class DwdBaseDb extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseDb().start(10019, 4, "dwd_base_db", Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 清洗 转换
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObj = JSONObject.parseObject(s);

                            // 提取数据
                            String type = jsonObj.getString("type");
                            String database = jsonObj.getString("database");
                            JSONObject dataObject = jsonObj.getJSONObject("data");

                            // 筛选数据
                            if (
                                    Constant.DW_DB.equals(database)
                                            &&
                                            (Constant.MAXWELL_TYPE_INSERT.equals(type) || Constant.MAXWELL_TYPE_UPDATE.equals(type))
                                            &&
                                            dataObject != null
                                            &&
                                            dataObject.size() >= 2
                            ) {
                                collector.collect(jsonObj);
                            }


                        } catch (Exception e) {
                            System.out.println(" 过滤掉脏数据:  " + s);
                        }
                    }
                }
        );
        // etlStream.print("ETL");


        // 读取配置表
        MySqlSource<String> mySqlSource = FlinkSourceUtil.getMySqlSource(Constant.GMALL_CONFIG_DB, Constant.TABLE_PROCESS_DWD);
        DataStreamSource<String> configDs = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource").setParallelism(1);
        

        // 配置流转换结构
        SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdDs = configDs.map(
                new MapFunction<String, TableProcessDwd>() {
                    @Override
                    public TableProcessDwd map(String value) throws Exception {
                        // r c u 取after的数据
                        // d 取before的数据
                        JSONObject jsonObj = JSON.parseObject(value);
                        String op = jsonObj.getString("op");
                        TableProcessDwd tableProcessDwd;
                        if (Constant.CDC_OP_D.equals(op)) {
                            // d 取before的数据
                            tableProcessDwd = jsonObj.getObject("before", TableProcessDwd.class);
                        } else {
                            // r c u 取after的数据
                            tableProcessDwd = jsonObj.getObject("after", TableProcessDwd.class);
                        }
                        // 补充op
                        tableProcessDwd.setOp(op);

                        return tableProcessDwd;
                    }
                }
        );
        // tableProcessDwdDs.print("TPD");


        // 将配置流处理成广播流  , connect , 处理广播数据 , 处理主流数据
        MapStateDescriptor<String, TableProcessDwd> tableProcessDwdStateDesc = new MapStateDescriptor<>("tableProcessDwdStateDesc", Types.STRING, Types.POJO(TableProcessDwd.class));
        BroadcastStream<TableProcessDwd> broadcastStream = tableProcessDwdDs.broadcast(tableProcessDwdStateDesc);

        etlStream.connect(broadcastStream)
                .process(
                        new BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>() {
                            @Override
                            public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
                                // 获取广播状态
                                ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = readOnlyContext.getBroadcastState(tableProcessDwdStateDesc);
                                // 拼接key
                                String key = value.getString("table") + ":" +value.getString("type");

                                // 从状态中获取对应的配置对象


                            }

                            @Override
                            public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context context, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
                                // 获取广播状态
                                BroadcastState<String, TableProcessDwd> broadcastState = context.getBroadcastState(tableProcessDwdStateDesc);
                                // 获取操作类型
                                String op = tableProcessDwd.getOp();
                                // 拼接key, 使用sourceTable 和 sourceType组合
                                String key = getKey(tableProcessDwd.getSourceTable(), tableProcessDwd.getSourceType());
                                if (Constant.CDC_OP_D.equals(op)){
                                    // 从状态中移除
                                    broadcastState.remove(key);
                                }else {
                                    // 更新状态
                                    broadcastState.put(key, tableProcessDwd);
                                }
                            }

                            private String getKey(String tableProcessDwd, String tableProcessDwd1) {
                                String key = tableProcessDwd + ":" + tableProcessDwd1;
                                return key;
                            }
                        }
                );


        // 写出搭配Kafka的不同主题中




    }
}*/
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TableProcessDwd;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author WEIYUNHUI
 * @date 2024/8/6 9:04
 *
 *  DWD事务事实表：
 *    流量域
 *         未经加工的事务事实表 -> 日志分流
 *
 *    交易域
 *         加购事务事实表
 *         下单事务事实表
 *         取消订单事务事实表
 *         支付成功事务事实表
 *         退单事务事实表
 *         退款成功事务事实表
 *
 *    互动域
 *         评论事务事实表
 *
 *         收藏商品事务事实表      favor_info -> insert
 *
 *    工具域
 *         优惠券领取事务事实表    coupon_user -> insert
 *         优惠券使用事务事实表    coupon_user -> update
 *
 *    用户域
 *         用户注册事务事实表     user_info -> insert
 *
 *
 */
public class DwdBaseDb extends BaseApp {

    public static void main(String[] args) {
        new DwdBaseDb().start(10019 , 4 , "dwd_base_db" , Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 清洗 转换
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);
        //etlStream.print("ETL");

        // 读取配置表
        SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdDs = readTableProcessDwd(env);

        // tableProcessDwdDs.print("TPD");

        // 将配置流处理成广播流  、 connect、 处理广播数据 、 处理主流数据

        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processDs = process(etlStream, tableProcessDwdDs);

        //processDs.print() ;

        // 写出到kafka的不同主题中

        writeToKafka(processDs);
    }

    private static void writeToKafka(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processDs) {
        processDs.sinkTo(FlinkSinkUtil.getKafkaSink()) ;
    }

    private static SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> process(SingleOutputStreamOperator<JSONObject> etlStream, SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdDs) {
        MapStateDescriptor<String, TableProcessDwd> tableProcessDwdStateDesc =
                new MapStateDescriptor<>("tableProcessDwdStateDesc", Types.STRING, Types.POJO(TableProcessDwd.class));
        BroadcastStream<TableProcessDwd> broadcastStream = tableProcessDwdDs.broadcast(tableProcessDwdStateDesc);

        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processDs = etlStream.connect(broadcastStream)
                .process(
                        new BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>() {

                            //定义Map,维护预加载的配置信息
                            Map<String, TableProcessDwd> configMap = new HashMap<>();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //预加载配置表
                                Connection connection = JdbcUtil.getConnection();

                                List<TableProcessDwd> tableProcessDwdList
                                        = JdbcUtil.queryList(
                                        connection,
                                        "select source_table, source_type , sink_table ,sink_columns from " + Constant.GMALL_CONFIG_DB + "." + Constant.TABLE_PROCESS_DWD,
                                        TableProcessDwd.class,
                                        true);

                                //封装到Map中
                                tableProcessDwdList.forEach(tableProcessDwd -> configMap.put(getKey(tableProcessDwd.getSourceTable(), tableProcessDwd.getSourceType()), tableProcessDwd));

                                JdbcUtil.closeConnection(connection);
                            }

                            @Override
                            public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDwd>> out) throws Exception {
                                //获取广播状态
                                ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = ctx.getBroadcastState(tableProcessDwdStateDesc);
                                //拼接key
                                String key = getKey(jsonObj.getString("table"), jsonObj.getString("type"));

                                //从状态中获取对应的配置对象
                                TableProcessDwd tableProcessDwd;
                                if (
                                        (tableProcessDwd = broadcastState.get(key)) != null
                                                ||
                                                (tableProcessDwd = configMap.get(key)) != null
                                ) {
                                    //提取data
                                    JSONObject dataJsonObj = jsonObj.getJSONObject("data");

                                    //提取需要的字段
                                    List<String> needColumns = Arrays.asList(tableProcessDwd.getSinkColumns().split(","));

                                    //移除掉不需要的字段
                                    dataJsonObj.keySet().removeIf(column -> !needColumns.contains(column));

                                    //写出数据
                                    out.collect(Tuple2.of(dataJsonObj, tableProcessDwd));
                                }

                            }

                            @Override
                            public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDwd>> out) throws Exception {
                                //获取广播状态
                                BroadcastState<String, TableProcessDwd> broadcastState = ctx.getBroadcastState(tableProcessDwdStateDesc);
                                //获取操作类型
                                String op = tableProcessDwd.getOp();
                                //拼接key ， 使用sourceTable 和 sourceType组合
                                String key = getKey(tableProcessDwd.getSourceTable(), tableProcessDwd.getSourceType());
                                if (Constant.CDC_OP_D.equals(op)) {
                                    //从状态中移除
                                    broadcastState.remove(key);
                                    //从Map中移除
                                    configMap.remove(key);
                                } else {
                                    //更新状态
                                    broadcastState.put(key, tableProcessDwd);
                                    //更新Map
                                    configMap.put(key, tableProcessDwd);
                                }
                            }

                            private String getKey(String tableName, String type) {

                                return tableName + ":" + type;
                            }
                        }
                );
        return processDs;
    }

    private static SingleOutputStreamOperator<TableProcessDwd> readTableProcessDwd(StreamExecutionEnvironment env) {
        MySqlSource<String> mysqlSource = FlinkSourceUtil.getMySqlSource(Constant.GMALL_CONFIG_DB, Constant.TABLE_PROCESS_DWD);
        DataStreamSource<String> configDs =
                env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource").setParallelism(1);

        // 配置流转换结构
        SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdDs = configDs.map(
                new MapFunction<String, TableProcessDwd>() {
                    @Override
                    public TableProcessDwd map(String value) throws Exception {

                        JSONObject jsonObj = JSON.parseObject(value);
                        String op = jsonObj.getString("op");
                        TableProcessDwd tableProcessDwd;
                        if (Constant.CDC_OP_D.equals(op)) {
                            // d: 取before数据
                            tableProcessDwd = jsonObj.getObject("before", TableProcessDwd.class);
                        } else {
                            // r c u :  取after数据
                            tableProcessDwd = jsonObj.getObject("after", TableProcessDwd.class);
                        }

                        //补充op
                        tableProcessDwd.setOp( op );
                        return tableProcessDwd;
                    }
                }
        ).setParallelism(1);
        return tableProcessDwdDs;
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(value);
                            String type = jsonObj.getString("type");
                            String database = jsonObj.getString("database");
                            JSONObject dataObject = jsonObj.getJSONObject("data");
                            //筛选数据
                            if (
                                    Constant.DW_DB.equals(database)
                                            &&
                                            (Constant.MAXWELL_TYPE_INSERT.equals(type) || Constant.MAXWELL_TYPE_UPDATE.equals(type))
                                            &&
                                            dataObject != null
                                            &&
                                            dataObject.size() >= 2
                            ) {
                                out.collect(jsonObj);
                            }

                        } catch (Exception e) {
                            System.out.println(" 过滤掉脏数据: " + value);
                        }
                    }
                }
        );
        return etlStream;
    }
}
